#include "mprpcprovider.h"
#include "mprpcapplication.h"
#include "rpcheader.pb.h"
#include "logger.h"
#include "zookeeperutil.h"
/*
这里是框架提供给外部使用的,可以发布rpc方法的函数接口
所以在设计上,参数一定要抽象化,而不是说为了给UserSerivce服务
参数就设计为UserService*对象,这样子就不是框架了
查看protobuf生成的.h文件，可以发现所有的message生成的类都继承了Service类
UserService : public fixbug::UserServiceRpc
UserServiceRpc : public google::protobuf::Service service
因此,我们可以将参数设置为Service*
*/
// 框架提供给外部使用的,可以发布rpc方法的函数接口
void RpcProvider::NotifyService(google::protobuf::Service* service)
{
    /*
    service_name ==>> ServiceInfo
    ServiceInfo ==>> service* 服务对象 + method_name
    method_name ==>> method服务方法
    */
   ServiceInfo service_info;

   // 获取服务对象的描述信息
   const google::protobuf::ServiceDescriptor* pserviceDesc = service->GetDescriptor();
   // 获取服务的名字
   std::string service_name = pserviceDesc->name();
   // 获取服务对象service的方法数量
   int methodCount = pserviceDesc->method_count();

   // Debug
   // std::cout << "service_name: " << service_name << std::endl;
   LOG_INFO("service_name: %s", service_name);

   // 存储服务名字--服务方法的hash
   for (int i = 0; i < methodCount; ++i) {
        // 获取服务对象指定下标的服务方法的描述
        const google::protobuf::MethodDescriptor* pmethodDesc = pserviceDesc->method(i);
        std::string method_name = pmethodDesc->name();
        service_info.m_methodMap.insert(std::make_pair(std::move(method_name), pmethodDesc));
        
        // Debug
        // std::cout << "method_name: " << method_name << std::endl;
        LOG_INFO("method_name: %s", method_name);
   }
   service_info.m_service = service;
   m_serviceMap.insert(std::make_pair(std::move(service_name), service_info));
}

// 启动rpc服务节点,开始提供rpc远程网络调用服务
void RpcProvider::Run()
{
    // 读取配置文件rpcserver的信息
    std::string ip = MprpcApplication::GetInstance().GetConfig().Load("rpcserverip");
    uint16_t port = atoi(MprpcApplication::GetInstance().GetConfig().Load("rpcserverport").c_str());
    muduo::net::InetAddress address(ip, port);

    // 创建TcpServer对象
    muduo::net::TcpServer server(&m_eventLoop, address, "RpcProvider");

    // 绑定连接回调函数和消息读写回调函数
    server.setConnectionCallback(std::bind(&RpcProvider::OnConnection, this, std::placeholders::_1));
    server.setMessageCallback(std::bind(&RpcProvider::OnMessage, this, std::placeholders::_1,
                                            std::placeholders::_2, std::placeholders::_3));

    // 设置网络层的线程数量
    server.setThreadNum(4);

    ZkClient zkCli;
    zkCli.Start();
    // service_name为永久节点   method_name为临时节点
    for (auto& service_name : m_serviceMap) {
        //   /service_name      eg:/UserServiceRpc
        std::string service_path = "/" + service_name.first;
        zkCli.Create(service_path.c_str(), nullptr, 0);
        for (auto& method_name : service_name.second.m_methodMap) {
            //  /service_name/method_name       eg:/UserServiceRpc/Login    存储当前这个rpc服务节点主机的ip和port
            std::string method_path = service_path + "/" + method_name.first;
            char method_path_data[128] = {0};
            sprintf(method_path_data, "%s:%d", ip.c_str(), port);
            // ZOO_EPHEMERAL表示znode是一个临时节点
            zkCli.Create(method_path.c_str(), method_path_data, strlen(method_path_data), ZOO_EPHEMERAL);
        }
    }

    // rpc服务端准备启动，打印信息
    std::cout << "RpcProvider start service at ip:" << ip << " port:" << port << std::endl;

    // 启动网络服务
    server.start();
    m_eventLoop.loop();
}

// 新的socket连接回调函数
void RpcProvider::OnConnection(const muduo::net::TcpConnectionPtr& conn)
{
    if (!conn->connected()) {
        // rpc服务端和rpc客户端断开连接了
        conn->shutdown();
    }
}

/*
在框架内部,RpcProvider和RpcConsumer协商好通信用的protobuf数据类型(即协议)
关注的对象:service_name method_name args  定义proto的message类型,进行数据头的序列化和反序列化
message: service_name method_name args_size
数据格式: header_size(4字节) + header_str + args_str, eg: 16UserServiceLoginwangzn123456
header_size以二进制方式描述header_str的数据大小
*/
// 已建立连接的用户的读写事件回调函数
void RpcProvider::OnMessage(const muduo::net::TcpConnectionPtr& conn, muduo::net::Buffer* buffer, muduo::Timestamp)
{
    // 网络层接受的远程rpc调用请求字符流
    std::string recv_buf = std::move(buffer->retrieveAllAsString());

    // 从字符流中读取前四个字节的内容,即header_size
    uint32_t header_size = 0;
    recv_buf.copy((char*)&header_size, 4, 0);

    // 根据header_size读取数据头的原始字符流,反序列化数据得到rpc请求的详细信息
    std::string rpc_header_str = recv_buf.substr(4, header_size);
    mprpc::RpcHeader rpcHeader;
    std::string service_name;
    std::string method_name;
    uint32_t args_size;
    // 反序列化
    if (rpcHeader.ParseFromString(rpc_header_str)) {
        // 反序列化成功
        service_name = rpcHeader.service_name();
        method_name = rpcHeader.method_name();
        args_size = rpcHeader.args_size();
    }
    else {
        // 反序列化失败
        std::cout << "rpc_header_str: " << rpc_header_str << " parse error!" << std::endl;
        return;
    }

    // 获取rpc方法参数的数据,即args
    std::string args_str = std::move(recv_buf.substr(4+header_size, args_size));

    // debug
    std::cout << "\n--------------------------------" << std::endl;
    std::cout << "recv_buf: " << recv_buf << std::endl;
    std::cout << "header_size: " << header_size << std::endl;
    std::cout << "rpc_header_str: " << rpc_header_str << std::endl;
    std::cout << "service_name: " << service_name << std::endl;
    std::cout << "method_name: " << method_name << std::endl;
    std::cout << "args_str: " << args_str << std::endl; 
    std::cout << "\n--------------------------------" << std::endl;

    // 获取service对象和method对象
    auto it = m_serviceMap.find(service_name);
    if (it == m_serviceMap.end()) {
        std::cout << service_name << " is not exist!" << std::endl;
        return;
    }
    auto mit = it->second.m_methodMap.find(method_name);
    if (mit == it->second.m_methodMap.end()) {
        std::cout << service_name << ":" << method_name << " is not exist!" << std::endl;
        return;
    }

    // 获取service对象,即new UserService()
    google::protobuf::Service* service = it->second.m_service;
    // 获取method方法的描述
    const google::protobuf::MethodDescriptor* method = mit->second;

    // 通过获取到的service对象和method方法,生成rpc方法调用的request请求参数和response响应参数
    google::protobuf::Message* request = service->GetRequestPrototype(method).New();
    if (!request->ParseFromString(args_str)) {
        std::cout << "request parse error, content: " << args_str << std::endl;
        return;
    }
    google::protobuf::Message* response = service->GetResponsePrototype(method).New();

    // 给下面的method方法的调用绑定一个Closure回调函数
    google::protobuf::Closure* done = google::protobuf::NewCallback<RpcProvider, 
                    const muduo::net::TcpConnectionPtr&, google::protobuf::Message*>
                    (this, &RpcProvider::SendRpcResponse, conn, response);
    
    // 在框架上根据远程rpc的请求,调用当前rpc节点上发布的方法
    // 即new UserService().Login(controller, request, response, done);
    service->CallMethod(method, nullptr, request, response, done);
}

// Closure的回调操作,用于序列化rpc的响应和网络发送
void RpcProvider::SendRpcResponse(const muduo::net::TcpConnectionPtr& conn, google::protobuf::Message* response)
{
    std::string response_str;
    if (response->SerializeToString(&response_str)) {
        // 对response进行序列化
        conn->send(response_str);
    }
    else {
        std::cout << "serialize response_str error!" << std::endl;
    }
    // 模拟http的短链接,rpcprovider主动断开连接
    conn->shutdown();
}